package mao.redis.jedisdelayqueue;

import mao.redis.jedispool.MyJedisPool;
import redis.clients.jedis.Jedis;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 使用redis进行延时任务的处理
 */

public class RedisDelayQueue {
    //使用threadLocal防止出现并发问题，但是何时关闭jedis连接是个问题。
    ThreadLocal<Jedis> client = new ThreadLocal<>();
    private String queue_name;

    public static class TaskItem{
        private String id;
        private java.lang.String msg;
    }


    public RedisDelayQueue(String name) {
        this.queue_name = name;
    }


    public String get(){
        Jedis client = this.client.get();
        if (client == null){
            this.client.set(MyJedisPool.getJedis());
        }
        client = this.client.get();

        //可以指定在特定socre范围的值。
        Set<String> set  = client.zrangeByScore(queue_name,0,System.currentTimeMillis(),0,1);
        if (!set.isEmpty()){
            String value = set.iterator().next();
            //删除成功说明当前线程抢到了这个任务。
            Long zrem = client.zrem(queue_name, value);
            if (zrem>0){
                return value;
            }else {
                return null;
            }
        }
        return null;
    }

    public Long put(String t){
        Jedis client = this.client.get();
        if (client == null){
            this.client.set(MyJedisPool.getJedis());
        }
        client = this.client.get();

        TaskItem msg_1 = new TaskItem();
        msg_1.msg = t;
        msg_1.id = UUID.randomUUID().toString();
        //定时5后再进行处理。
        Long zadd = client.zadd(queue_name, Double.valueOf(System.currentTimeMillis())+5000L, t);
        return zadd;
    }

    public static void main(String[] args) {
        final RedisDelayQueue queue = new RedisDelayQueue("q_demo");
        Consumer con = new Consumer(queue);
        Producer pro = new Producer(queue);

        ExecutorService exec = Executors.newCachedThreadPool();
        for (int i = 0; i < 10; i++) {
            exec.execute(con);
        }
        for (int i = 0; i < 10; i++) {
            exec.execute(pro);
        }

        try {
            Thread.sleep(8000);
            exec.shutdownNow();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}
